package com.fc.connector.source;

import org.apache.flink.api.connector.source.*;
import org.apache.flink.core.io.SimpleVersionedSerializer;
import org.apache.flink.table.data.RowData;

import java.io.IOException;

/**
 * https://blog.csdn.net/MCC_MCC_MCC/article/details/143781789
 *Source API 是一个工厂模式的接口，用于创建以下组件。
 * • Split Enumerator
 * • Source Reader
 * • Split Serializer
 * • Enumerator Checkpoint Serializer
 * 除此之外，Source 还提供了 Boundedness 的特性，从而使得 Flink 可以选择合适的模式来运行 Flink 任务。
 * Source 实现应该是可序列化的，因为 Source 实例会在运行时被序列化并上传到 Flink 集群。
 *
 */
public class MySource implements Source<RowData, MySplit, MySplitEnumerator.CheckpointState> {
    @Override
    public Boundedness getBoundedness() {
        return Boundedness.CONTINUOUS_UNBOUNDED;
    }


    /**
     * 创建枚举器
     * @param context The {@link SplitEnumeratorContext context} for the split enumerator.
     * @return
     */
    @Override
    public SplitEnumerator<MySplit, MySplitEnumerator.CheckpointState> createEnumerator(SplitEnumeratorContext<MySplit> context) {
        return new MySplitEnumerator(context, null);

    }

    @Override
    public SplitEnumerator<MySplit, MySplitEnumerator.CheckpointState> restoreEnumerator(
            SplitEnumeratorContext<MySplit> context, MySplitEnumerator.CheckpointState checkpoint) {
        return new MySplitEnumerator(context, checkpoint);
    }

    @Override
    public SourceReader<RowData, MySplit> createReader(SourceReaderContext sourceReaderContext) throws Exception {
        return new MySourceReader(sourceReaderContext);
    }


    @Override
    public SimpleVersionedSerializer<MySplit> getSplitSerializer() {
        System.out.println("getSplitSerializer-------------------------");
        return new SimpleVersionedSerializer<MySplit>() {
            @Override
            public int getVersion() {
                return 0;
            }

            @Override
            public byte[] serialize(MySplit obj) throws IOException {
                return new byte[0];
            }

            @Override
            public MySplit deserialize(int version, byte[] serialized) throws IOException {
                return new MySplit("1");
            }
        };
    }

    @Override
    public SimpleVersionedSerializer<MySplitEnumerator.CheckpointState> getEnumeratorCheckpointSerializer() {
        System.out.println("getEnumeratorCheckpointSerializer-------------------------");
        return new MySplitEnumerator.CheckpointSerializer();
    }

}